#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include <time.h>
#include <map>
#include <iostream>
#include <sstream>

#include "raptor-lite/raptor-lite.h"

struct echo_message {
    uint32_t payload_size;
    uint32_t request_id;
    int64_t millseconds;
};

class HttpEchoServer : public raptor::AcceptorHandler,
                       public raptor::MessageHandler,
                       public raptor::ProtocolHandler,
                       public raptor::EndpointClosedHandler {
public:
    HttpEchoServer(/* args */);
    ~HttpEchoServer();

    void OnAccept(raptor::Endpoint ep, raptor::Property &settings) {
        settings({{"SocketRecvTimeoutMs", 5000}, {"SocketSendTimeoutMs", 5000}});
        log_debug("OnAccept: %s, fd:%d", ep.PeerString().c_str(), ep.SocketFd());
        container->AttachEndpoint(ep);
    }

    int OnCheckPackageLength(raptor::Endpoint ep, const void *data, size_t len) {
        // log_debug("OnCheckPackageLength:%u", len);
        (void)ep;
        if (len < 4) return 0;
        auto pack_len = *reinterpret_cast<const uint32_t *>(data);
        return static_cast<int>(pack_len);
    }

    void OnMessage(raptor::Endpoint ep, const raptor::Slice &msg) {
        // auto request = reinterpret_cast<const echo_message *>(msg.begin());
        // log_debug("OnMessage: req=%lu millsecs=%lld", request->request_id, request->millseconds);
        ep.SendMsg(msg);
    }

    void OnClosed(raptor::Endpoint ep, raptor_error desc) {
        if (desc == RAPTOR_ERROR_NONE) {
            log_debug("OnClosed: %s", ep.PeerString().c_str());
        } else {
            log_debug("OnClosed: %s %s", ep.PeerString().c_str(), desc->ToString().c_str());
        }
    }

    void init() {

        uint32_t cpus = raptor::GetNumberofCPUCores();

        engine = RaptorCreateEngine();

        raptor::Property p{
            {raptor::config::kAcceptorHandler, static_cast<raptor::AcceptorHandler *>(this)},
            {raptor::config::kMessageHandler, static_cast<raptor::MessageHandler *>(this)},
            {raptor::config::kProtocolHandler, static_cast<raptor::ProtocolHandler *>(this)},
            {raptor::config::kEndpointClosedHandler,
             static_cast<raptor::EndpointClosedHandler *>(this)},
            {raptor::config::kCheckReceivingInterval, false},
            {raptor::config::kRecvSendThreads, cpus}};

        raptor_error err = engine->CreateContainer(p, &container);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create container: %s", err->ToString().c_str());
            abort();
        }
        err = engine->CreateAcceptor(p, &acceptor);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create acceptor: %s", err->ToString().c_str());
            abort();
        }
    }

    void start_and_listening(const std::string &addr) {
        raptor_error err = acceptor->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start acceptor: %s", err->ToString().c_str());
            return;
        }
        err = container->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to start container: %s", err->ToString().c_str());
            return;
        }
        err = acceptor->AddListening(addr);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to listening %s, error: %s", addr.c_str(), err->ToString().c_str());
            return;
        }
    }

    void stop() {
        acceptor->Shutdown();
        container->Shutdown();
    }

private:
    RaptorEngine engine;
    raptor::Container *container = nullptr;
    raptor::Acceptor *acceptor   = nullptr;
};

HttpEchoServer::HttpEchoServer(/* args */) {
    // Do not call the raptor::Create function here, because the current
    // class (inherited from XXHandler) has not been constructed.
}

HttpEchoServer::~HttpEchoServer() {
    engine->DestroyAcceptor(acceptor);
    engine->DestroyContainer(container);
}

int main() {
    printf(" ---- prepare start echo server ---- \n");
    RaptorGlobalStartup();
    HttpEchoServer server;
    server.init();
    server.start_and_listening("localhost:50051");
    log_warn("Press any key to quit");
    getchar();
    server.stop();
    RaptorGlobalCleanup();
    return 0;
}
